package com.yanggu.flink.datastream_api.multi_stream_transform.split_stream

import com.yanggu.flink.datastream_api.pojo.SensorReading
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 侧输出流功能, 可以给数据打上一个tag, 在后面的过程中可以根据tag进行获取
 * split和select算子
 * split => 把DataStream => SplitStream 根据某些特征把一个DataStream拆分成两个或者多个DataStream
 * select => 从一个SplitStream中获取一个或者多个DataStream
 * 高版本的flink已经删除了该api, 可以使用sideOutput侧输出流进行代替
 */
object SplitBySideOutputDemo {

  /**
   * 需求描述: 传感器数据按照温度高低(30℃), 拆分成三个流
   * 大于30℃、等于30℃、等于30℃
   */
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    val stream = environment.fromCollection(List(
      SensorReading("sensor_1", System.currentTimeMillis(), 25.8D),
      SensorReading("sensor_2", System.currentTimeMillis(), 45.8D),
      SensorReading("sensor_3", System.currentTimeMillis(), 65.8D),
      SensorReading("sensor_4", System.currentTimeMillis(), 30.0D)
    ))

    //高版本的flink删除了split算子, 取而代之的是OutputTag类和getSideOutput(outputTag)算子
    val gt30 = OutputTag[SensorReading]("gt30")
    val le30 = OutputTag[SensorReading]("le30")

    val outputTagStream = stream
      //可以使用process方法, 需要传入ProcessFunction方法
      //总之在可以使用上下文的api处都可以进行侧输出流
      .process((value: SensorReading,
                ctx: ProcessFunction[SensorReading, SensorReading]#Context,
                out: Collector[SensorReading]) => {
        //传感器数据按照温度高低(30℃), 拆分成三个流
        if (value.temperature > 30.0D) {
          //大于30℃的
          ctx.output(gt30, SensorReading(value.id, value.timestamp, value.temperature))
        } else if (value.temperature < 30.0D) {
          //小于30℃
          ctx.output(le30, value)
        } else {
          //等于30℃
          out.collect(value)
        }
      })

    //根据OutputTag获取流
    val gt30Stream = outputTagStream.getSideOutput(gt30)
    val le30Stream = outputTagStream.getSideOutput(le30)

    //分别打印30℃和小于等于30℃的流
    gt30Stream.print("大于30℃").setParallelism(1)
    le30Stream.print("小于30℃").setParallelism(1)

    //打印原有的流
    outputTagStream.print("等于30℃").setParallelism(1)

    environment.execute()
  }

}
